﻿/*
-----------------------------------------------------------------------------
This source file is part of Cell Cloud.

Copyright (c) 2009-2012 Cell Cloud Team - cellcloudproject@gmail.com

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
-----------------------------------------------------------------------------
*/

#include "CellStableHeaders.h"
#include "CellExecutorListener.h"
#include "CellScheduledThreadPoolExecutor.h"
#include "CellTimeUnit.h"
#include "CellUtil.h"
#include "CellLoggerManager.h"

namespace cell
{

//-----------------------------------------------------------------------
ScheduledThreadPoolExecutor::ScheduledThreadPoolExecutor(const String& name,
	uint32 corePoolSize, uint32 maxPoolSize)
	: ThreadPoolExecutor(name, corePoolSize, maxPoolSize)
{
	mPolicy = CELL_NEW ScheduledHalfWindowPolicy(this);
	mPolicy->start();
}
//-----------------------------------------------------------------------
ScheduledThreadPoolExecutor::~ScheduledThreadPoolExecutor(void)
{
	std::vector<Runnable*> taskList;
	ScheduledFutureMap::iterator iter, end = mScheduledFutures.end();
	for (iter = mScheduledFutures.begin(); iter != end; ++iter)
	{
		ScheduledFutureImpl* future = iter->second;
		taskList.push_back(future->mRunnable);
	}

	std::vector<Runnable*>::iterator titer;
	for (titer = taskList.begin(); titer != taskList.end(); ++titer)
	{
		remove(*titer);
	}

	awaitExecution();

	if (NULL != mPolicy)
	{
		mPolicy->stop();
		CELL_DELETE mPolicy;
		mPolicy = NULL;
	}

	end = mScheduledFutures.end();
	for (iter = mScheduledFutures.begin(); iter != end; ++iter)
	{
		ScheduledFutureImpl* future = iter->second;
		CELL_DELETE future;
	}

	destroyDesertedFutures();
}
//-----------------------------------------------------------------------
ScheduledFuture* ScheduledThreadPoolExecutor::schedule(
	Runnable* task, ulong delay, uint8 unit)
{
	ScheduledFutureImpl* future = NULL;

	{
		CELL_LOCK_MUTEX(mMutex);

		ScheduledFutureMap::const_iterator iter = mScheduledFutures.find(task);
		if (iter != mScheduledFutures.end())
		{
			future = iter->second;
		}
		else
		{
			if (mScheduledFutures.size() >= getMaxPoolSize())
			{
				char log[128] = {0x0};
				sprintf(log, "[CC] ScheduledThreadPoolExecutor task number reaches the maximum : %d", mScheduledFutures.size());
				LoggerManager::getSingleton().log(log);
				return NULL;
			}

			// Create the schedule future which task in executing list
			future = CELL_NEW ScheduledFutureImpl(this, task);
			mScheduledFutures.insert(ScheduledFutureMap::value_type(task, future));
		}
	}

	mPolicy->removeTask(task);
	mPolicy->addTask(task, delay, unit);

	return future;
}
//-----------------------------------------------------------------------
ScheduledFuture* ScheduledThreadPoolExecutor::scheduleAtFixedRate(
	Runnable* task, ulong initialDelay, ulong period, uint8 unit)
{
	ScheduledFutureImpl* future = NULL;

	{
		CELL_LOCK_MUTEX(mMutex);

		ScheduledFutureMap::const_iterator iter = mScheduledFutures.find(task);
		if (iter != mScheduledFutures.end())
		{
			future = iter->second;
		}
		else
		{
			if (mScheduledFutures.size() >= getMaxPoolSize())
			{
				char log[128] = {0x0};
				sprintf(log, "[CC] ScheduledThreadPoolExecutor task number reaches the maximum : %d", mScheduledFutures.size());
				LoggerManager::getSingleton().log(log);
				return NULL;
			}

			future = CELL_NEW ScheduledFutureImpl(this, task);
			mScheduledFutures.insert(ScheduledFutureMap::value_type(task, future));
		}
	}

	mPolicy->removeTask(task);
	mPolicy->addTaskAtFixedRate(task, initialDelay, period, unit);

	return future;
}
//-----------------------------------------------------------------------
ScheduledFuture* ScheduledThreadPoolExecutor::scheduleWithFixedDelay(
	Runnable* task, ulong initialDelay, ulong delay, uint8 unit)
{
	ScheduledFutureImpl* future = NULL;

	{
		CELL_LOCK_MUTEX(mMutex);

		ScheduledFutureMap::const_iterator iter = mScheduledFutures.find(task);
		if (iter != mScheduledFutures.end())
		{
			future = iter->second;
		}
		else
		{
			if (mScheduledFutures.size() >= getMaxPoolSize())
			{
				char log[128] = {0x0};
				sprintf(log, "[CC] ScheduledThreadPoolExecutor task number reaches the maximum : %d", mScheduledFutures.size());
				LoggerManager::getSingleton().log(log);
				return NULL;
			}

			future = CELL_NEW ScheduledFutureImpl(this, task);
			mScheduledFutures.insert(ScheduledFutureMap::value_type(task, future));
		}
	}

	mPolicy->removeTask(task);
	mPolicy->addTaskWithFixedDelay(task, initialDelay, delay, unit);

	return future;
}
//-----------------------------------------------------------------------
void ScheduledThreadPoolExecutor::remove(Runnable* task)
{
	mPolicy->removeTask(task);

	CELL_LOCK_MUTEX(mMutex);

	ScheduledFutureMap::iterator iter = mScheduledFutures.find(task);
	if (iter != mScheduledFutures.end())
	{
		ScheduledFutureList::iterator fiter = std::find(mDesertedFutures.begin(), mDesertedFutures.end(), iter->second);
		if (fiter == mDesertedFutures.end())
			mDesertedFutures.push_back(iter->second);

		mScheduledFutures.erase(iter);
	}
}
//-----------------------------------------------------------------------
int ScheduledThreadPoolExecutor::getTaskCount(void)
{
	CELL_LOCK_MUTEX(mMutex);
	return mScheduledFutures.size();
}
//-----------------------------------------------------------------------
void ScheduledThreadPoolExecutor::cancel(Runnable* task)
{
	mPolicy->removeTask(task);
}
//-----------------------------------------------------------------------
void ScheduledThreadPoolExecutor::addListener(ExecutorListener* listener)
{
	ThreadPoolExecutor::addListener(listener);
}
//-----------------------------------------------------------------------
void ScheduledThreadPoolExecutor::removeListener(ExecutorListener* listener)
{
	ThreadPoolExecutor::removeListener(listener);
}
//-----------------------------------------------------------------------
bool ScheduledThreadPoolExecutor::execute(Runnable* task)
{
	return ThreadPoolExecutor::execute(task);
}
//-----------------------------------------------------------------------
void ScheduledThreadPoolExecutor::beforeExecute(Runnable* task)
{
	mPolicy->fireBeforeExecute(task);
}
//-----------------------------------------------------------------------
void ScheduledThreadPoolExecutor::afterExecute(Runnable* task)
{
	mPolicy->fireAfterExecute(task);
}
//-----------------------------------------------------------------------
bool ScheduledThreadPoolExecutor::isCancelled(Runnable* task)
{
	return mPolicy->hasTask(task) == true ? false : true;
}
//-----------------------------------------------------------------------
void ScheduledThreadPoolExecutor::destroyDesertedFutures(void)
{
	CELL_LOCK_MUTEX(mMutex);

	if (mDesertedFutures.empty())
		return;

	ScheduledFutureList::iterator iter, end = mDesertedFutures.end();
	for (iter = mDesertedFutures.begin(); iter != end; ++iter)
	{
		ScheduledFutureImpl* sf = (*iter);
		CELL_DELETE sf;
	}

	mDesertedFutures.clear();
}



//-----------------------------------------------------------------------
ScheduledHalfWindowPolicy::ScheduledHalfWindowPolicy(
	ScheduledThreadPoolExecutor* executor)
	: mWorking(false)
	, mSpinning(false)
	, mExecutor(executor)
	, mRollWindowSizeMillis(10)
{
}
//-----------------------------------------------------------------------
ScheduledHalfWindowPolicy::~ScheduledHalfWindowPolicy(void)
{
	CELL_LOCK_MUTEX(mMoniter);

	ScheduleMap::iterator iter, end = mSchedules.end();
	for (iter = mSchedules.begin(); iter != end; ++iter)
	{
		Schedule* sch = iter->second;
		CELL_DELETE sch;
	}
}
//-----------------------------------------------------------------------
void ScheduledHalfWindowPolicy::run(void)
{
	uint64 halfWindowSize = mRollWindowSizeMillis * 0.5;
	uint64 current = Util::currentTimeMillis();
	uint64 frontWindow = current - halfWindowSize;
	uint64 backWindow = current + halfWindowSize;

	ScheduleMap::iterator iter, end;
	ScheduleList::iterator diter, dend;

	mSpinning = true;
	mWorking = true;
	while (mSpinning)
	{
		{
			CELL_LOCK_MUTEX(mMoniter);

			end = mSchedules.end();
			for (iter = mSchedules.begin(); iter != end; ++iter)
			{
				Schedule* sch = iter->second;
				if ((sch->delayMillis > 0 || (sch->delayMillis == 0 && sch->periodMillis == 0)) && sch->executing)
					continue;

				// Calculate task time window:
				// The A, B and C are task.
				// The node @ is the executing time node.
				// 
				// [-A-----A-----A-----A-----A-----A-----A-----A-----A-----] -> 9
				// [------B--------B--------B--------B--------B--------B---] -> 6
				// [----------C------------C------------C------------C-----] -> 4
				// [---@-----@-----@-----@-----@-----@-----@-----@-----@---]
				//  |*****|*****|*****|*****|*****|*****|*****|*****|*****|
				//     A   A,B,C  A,B   A,C   A,B   A,B   A,C   A,B  A,B,C   -> 9,6,4
				//
				// Task A had executed 9 times, it is correct.
				// Task B had executed 6 times, it is correct.
				// Task C had executed 4 times, it is correct.
				//
				// Deviation analysis:
				//
				// [-A-----A-----A-----A-----A-----A-----A-----A-----A-----X] -> 9
				// [------B--------B--------B--------B--------B--------B----] -> 6
				// [----------C------------C------------C------------C------] -> 4
				// [---@---**--@-----@--*---@--*---@-----@---**--@---*--@---]
				//  |*****| |*****|*****||*****||*****|*****| |*****||*****|
				//     A      A,C   A,B   A,B,C   A,B   A,C     A,B   A,B,C   -> 8,5,4
				//  |*****|*******|*****|******|******|*****|*******|******| 
				//     A    A,B,C   A,B   A,B,C   A,B   A,C    A,B    A,B,C   -> 8,6,4
				//            A?                                              -> 1(A)
				//
				// So task delay window size must greater than roll window size.
				//
				bool executable = ((sch->permitTime >= frontWindow) && (sch->permitTime < backWindow));
				if (executable)
				{
					// Execute the task which in schedule
					bool state = mExecutor->execute(sch->task);
					if (false == state)
					{
						char log[64] = {0x0};
						sprintf(log, "[CC] 任务表 %s 执行任务失败", mExecutor->getName().c_str());
						LoggerManager::getSingleton().log(log);
					}
				}
			}

			// 处理被遗弃的 Schedule
			if (false == mDesertedSchedules.empty())
			{
				ScheduleList removeList;

				dend = mDesertedSchedules.end();
				for (diter = mDesertedSchedules.begin(); diter != dend; ++diter)
				{
					Schedule* sch = (*diter);
					if (sch->executing)
						continue;

					if (sch->deserted)
					{
						removeList.push_back(sch);
					}
					else
					{
						sch->deserted = true;
					}
				}

				dend = removeList.end();
				ScheduleList::iterator fiter;
				for (diter = removeList.begin(); diter != dend; ++diter)
				{
					Schedule* sch = (*diter);

					fiter = std::find(mDesertedSchedules.begin(), mDesertedSchedules.end(), sch);
					if (fiter != mDesertedSchedules.end())
					{
						mDesertedSchedules.erase(fiter);
					}

					CELL_DELETE sch;
				}

				removeList.clear();
			}
		}

		// 处理执行器废弃的 Futures
		mExecutor->destroyDesertedFutures();

		// Sleep
		CELL_THREAD_SLEEP(mRollWindowSizeMillis);

		current = Util::currentTimeMillis();
		frontWindow = backWindow;
		backWindow = current + halfWindowSize;
	}

	// 结束
	mWorking = false;
}
//-----------------------------------------------------------------------
void ScheduledHalfWindowPolicy::stop(void)
{
	{
		CELL_LOCK_MUTEX(mMoniter);

		ScheduleMap::iterator iter, end = mSchedules.end();
		for (iter = mSchedules.begin(); iter != end; ++iter)
		{
			Schedule* sch = iter->second;

			if (sch->executing == false)
				sch->deserted = true;

			ScheduleList::iterator siter = std::find(mDesertedSchedules.begin(), mDesertedSchedules.end(), sch);
			if (siter == mDesertedSchedules.end())
				mDesertedSchedules.push_back(sch);
		}

		mSchedules.clear();
	}

	mSpinning = false;

	while (mWorking)
	{
		CELL_THREAD_SLEEP(10);
	}

	ScheduleList scheduleList;

	{
		CELL_LOCK_MUTEX(mMoniter);

		ScheduleList::iterator iter, end = mDesertedSchedules.end();
		for (iter = mDesertedSchedules.begin(); iter != end; ++iter)
		{
			(*iter)->deserted = true;
			scheduleList.push_back((*iter));
		}

		mDesertedSchedules.clear();
	}

	CELL_THREAD_SLEEP(0);

	ScheduleList::iterator iter, end = scheduleList.end();
	for (iter = scheduleList.begin(); iter != end; ++iter)
	{
		CELL_DELETE (*iter);
	}
}
//-----------------------------------------------------------------------
ScheduledHalfWindowPolicy::Schedule* ScheduledHalfWindowPolicy::addTask(Runnable* task,
	ulong initialDelay, uint8 unit)
{
	CELL_LOCK_MUTEX(mMoniter);

	Schedule* sch = CELL_NEW Schedule(task);
	sch->initialDelayMillis = TimeUnit::convertMillisecond(initialDelay, unit);
	if (sch->initialDelayMillis <= mRollWindowSizeMillis)
		sch->initialDelayMillis = mRollWindowSizeMillis + mRollWindowSizeMillis * 0.2;
	sch->permitTime = sch->origin + sch->initialDelayMillis;

	mSchedules.insert(ScheduleMap::value_type(task, sch));
	return sch;
}
//-----------------------------------------------------------------------
ScheduledHalfWindowPolicy::Schedule* ScheduledHalfWindowPolicy::addTaskAtFixedRate(
	Runnable* task, ulong initialDelay, ulong period, uint8 unit)
{
	CELL_LOCK_MUTEX(mMoniter);

	Schedule* sch = CELL_NEW Schedule(task);
	sch->initialDelayMillis = TimeUnit::convertMillisecond(initialDelay, unit);
	if (sch->initialDelayMillis <= mRollWindowSizeMillis)
		sch->initialDelayMillis = mRollWindowSizeMillis + mRollWindowSizeMillis * 0.2;
	sch->permitTime = sch->origin + sch->initialDelayMillis;
	// Sets period millisecond
	sch->periodMillis = TimeUnit::convertMillisecond(period, unit);
	if (sch->periodMillis < mRollWindowSizeMillis)
		sch->periodMillis = mRollWindowSizeMillis;

	mSchedules.insert(ScheduleMap::value_type(task, sch));
	return sch;
}
//-----------------------------------------------------------------------
ScheduledHalfWindowPolicy::Schedule* ScheduledHalfWindowPolicy::addTaskWithFixedDelay(
	Runnable* task, ulong initialDelay, ulong delay, uint8 unit)
{
	CELL_LOCK_MUTEX(mMoniter);

	Schedule* sch = CELL_NEW Schedule(task);
	sch->initialDelayMillis = TimeUnit::convertMillisecond(initialDelay, unit);
	if (sch->initialDelayMillis <= mRollWindowSizeMillis)
		sch->initialDelayMillis = mRollWindowSizeMillis + mRollWindowSizeMillis * 0.2;
	sch->permitTime = sch->origin + sch->initialDelayMillis;
	// Sets delay millisecond
	sch->delayMillis = TimeUnit::convertMillisecond(delay, unit);
	if (sch->delayMillis < mRollWindowSizeMillis)
		sch->delayMillis = mRollWindowSizeMillis;

	mSchedules.insert(ScheduleMap::value_type(task, sch));
	return sch;
}
//-----------------------------------------------------------------------
void ScheduledHalfWindowPolicy::removeTask(Runnable* task)
{
	CELL_LOCK_MUTEX(mMoniter);

	ScheduleMap::iterator iter = mSchedules.find(task);
	if (iter != mSchedules.end())
	{
		Schedule* sch = iter->second;

		ScheduleList::iterator siter = std::find(mDesertedSchedules.begin(), mDesertedSchedules.end(), sch);
		if (siter == mDesertedSchedules.end())
			mDesertedSchedules.push_back(sch);

		mSchedules.erase(iter);
	}
}
//-----------------------------------------------------------------------
bool ScheduledHalfWindowPolicy::hasTask(Runnable* task)
{
	CELL_LOCK_MUTEX(mMoniter);

	ScheduleMap::const_iterator iter = mSchedules.find(task);
	if (iter != mSchedules.end())
	{
		return true;
	}
	return false;
}
//-----------------------------------------------------------------------
void ScheduledHalfWindowPolicy::fireBeforeExecute(Runnable* task)
{
	CELL_LOCK_MUTEX(mMoniter);

	ScheduleMap::const_iterator iter = mSchedules.find(task);
	if (iter != mSchedules.end())
	{
		Schedule* sch = iter->second;
		sch->lastBeginTime = Util::currentTimeMillis();
		if (sch->periodMillis > 0)
		{
			sch->permitTime = sch->lastBeginTime + sch->periodMillis;
		}
		else if (sch->delayMillis > 0)
		{
			sch->permitTime = 0;
		}
		sch->executing = true;
	}
}
//-----------------------------------------------------------------------
void ScheduledHalfWindowPolicy::fireAfterExecute(Runnable* task)
{
	bool cancel = false;

	{
		CELL_LOCK_MUTEX(mMoniter);

		ScheduleMap::const_iterator iter = mSchedules.find(task);
		if (iter != mSchedules.end())
		{
			Schedule* sch = iter->second;
			sch->lastEndTime = Util::currentTimeMillis();
			if (sch->delayMillis > 0)
			{
				sch->permitTime = sch->lastEndTime + sch->delayMillis;
			}
			else if (sch->periodMillis > 0)
			{
				// Nothing
			}
			else
			{
				// Cancel
				cancel = true;
			}

			sch->executing = false;
			sch->worked = true;
		}
	}

	if (cancel)
		mExecutor->cancel(task);
}

//-----------------------------------------------------------------------
ScheduledHalfWindowPolicy::Schedule::Schedule(Runnable* t)
	: task(t)
	, worked(false)
	, deserted(false)
	, executing(false)
	, periodMillis(0)
	, delayMillis(0)
	, lastBeginTime(0)
	, lastEndTime(0)
	, initialDelayMillis(0)
{
	origin = Util::currentTimeMillis();
	permitTime = origin;
}
//-----------------------------------------------------------------------
ScheduledHalfWindowPolicy::Schedule::~Schedule(void)
{
}




//-----------------------------------------------------------------------
ScheduledFutureImpl::ScheduledFutureImpl(
	ScheduledThreadPoolExecutor* executor, Runnable* runnable)
	: mExecutor(executor)
	, mRunnable(runnable)
{
	assert(mRunnable);
}
//-----------------------------------------------------------------------
ScheduledFutureImpl::~ScheduledFutureImpl(void)
{
	mExecutor = NULL;
	mRunnable = NULL;
}
//-----------------------------------------------------------------------
ulong ScheduledFutureImpl::getDelay(uint8 unit)
{
	return 0;
}
//-----------------------------------------------------------------------
bool ScheduledFutureImpl::cancel(void)
{
	mExecutor->cancel(mRunnable);
	return true;
}
//-----------------------------------------------------------------------
inline bool ScheduledFutureImpl::isCancelled(void)
{
	return mExecutor->isCancelled(mRunnable);
}

} // end namespace cell
